package org.framework.gofun.mq;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;


/**
 * Consumer，订阅消息
 */
public class Consumer{

	private  Logger log = LoggerFactory.getLogger(Consumer.class);
	private DefaultMQPushConsumer consumer;
	private String groupName = "group_gofun";// group_gofun
	private String namesrvAddr = "192.168.146.128:9876";// 
	private String topicName = "topic_gofun_api";// topic_gofun
	private String consumerGroup = "consumer_group_api";
	private String tag = "*";// tag
	private MessageListenerConcurrently messageListener;
	
	public static void main(String[] args) {
		Consumer consumer = new Consumer();
		consumer.init();
	}
	
	// 初始化方法
	public void init() {

		consumer = new DefaultMQPushConsumer(groupName);
		if (namesrvAddr != null) {
			consumer.setNamesrvAddr(namesrvAddr);
		}
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
		consumer.setConsumerGroup(consumerGroup);
		try {
			consumer.subscribe(topicName, tag);
			consumer.registerMessageListener(messageListener);
			consumer.start();
			log.info("namesrvAddr="+namesrvAddr+",groupName="+groupName+" 消息队列初始化完毕...");
		} catch (MQClientException e) {
			log.error("namesrvAddr="+namesrvAddr+",groupName="+groupName+" 消息队列初始化失败...",e);
		}
	}
	
	public void shutdown() {
        	consumer.shutdown();
			log.info(groupName+" 消息队列shutdown完毕...");
	}
	
	public String getGroupName() {
		return groupName;
	}

	public void setGroupName(String groupName) {
		this.groupName = groupName;
	}

	public String getNamesrvAddr() {
		return namesrvAddr;
	}

	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}

	public String getTopicName() {
		return topicName;
	}

	public void setTopicName(String topicName) {
		this.topicName = topicName;
	}

	public MessageListenerConcurrently getMessageListener() {
		return messageListener;
	}

	public void setMessageListener(MessageListenerConcurrently messageListener) {
		this.messageListener = messageListener;
	}

	public void setTag(String tag) {
		this.tag = tag;
	}

	public String getConsumerGroup() {
		return consumerGroup;
	}

	public void setConsumerGroup(String consumerGroup) {
		this.consumerGroup = consumerGroup;
	}
}
